package cn.itcast.streaming.sink;

import cn.itcast.streaming.entity.ItcastDataObj;
import cn.itcast.streaming.utils.DateUtil;
import cn.itcast.streaming.utils.StringUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Author itcast
 * Date 2021/9/22 11:28
 * Desc 将计算指标使用到的部分车辆数据保存到hbase中
 * 首先将数据保存到 BufferMutable 缓存对象中，然后flush到hbase中
 * 开发步骤：
 * 1.重写open方法，连接hbase数据表连接，获取连接串url，用户名密码，表
 * 2.实现invoke方法，将每条数据写入到hbase中，拼 put
 * 3.关闭close方法，关闭连接
 */
public class SrcDetailToHBaseOptimizerSink extends RichSinkFunction<ItcastDataObj> {
    private static final Logger logger = LoggerFactory.getLogger(SrcDetailToHBaseOptimizerSink.class);

    //定义当前类的私有变量
    private String tableName;
    //定义创建连接
    Connection conn = null;
    //定义表操作的对象
    BufferedMutator mutator = null;

    //2.创建一个有参数-表名的构造方法
    public SrcDetailToHBaseOptimizerSink(String _tableName) {
        this.tableName = _tableName;
    }

    //3.重写open方法
    @Override
    public void open(Configuration parameters) throws Exception {
        //3.1 从上下文获取到全局的参数
        ParameterTool tools = (ParameterTool) getRuntimeContext()
                .getExecutionConfig()
                .getGlobalJobParameters();
        //3.2 设置hbase的配置，Zookeeper Quorum集群和端口和TableInputFormat的输入表
        String zkQuorum = tools.getRequired("zookeeper.quorum");
        String port = tools.getRequired("zookeeper.clientPort");
        org.apache.hadoop.conf.Configuration def = new org.apache.hadoop.conf.Configuration();
        def.set(HConstants.ZOOKEEPER_QUORUM, zkQuorum);
        def.set(HConstants.ZOOKEEPER_CLIENT_PORT, port);
        def.set(TableInputFormat.INPUT_TABLE, tableName);

        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create(def);
        //3.3 通过连接工厂创建连接
        conn = ConnectionFactory.createConnection(conf);
        //todo 设置缓存对象的多大、多长时间刷新到hbase到
        BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf(tableName));
        params.writeBufferSize(10 * 1024 * 1024L);
        params.setWriteBufferPeriodicFlushTimeoutMs(5 * 1000L);
        //3.4 通过连接获取表对象
        try {
            mutator = conn.getBufferedMutator(params);
        } catch (Exception ex) {
            logger.error("当前获取bufferedMutator 失败：" + ex.getMessage());
        }
    }

    //5. 重写 invoke 方法，将读取的数据写入到 hbase
    @Override
    public void invoke(ItcastDataObj value, Context context) throws Exception {
        //5.1 setDataSourcePut输入参数value，返回put对象
        try {
            Put put = setDataSourcePut(value);
            mutator.mutate(put);
            //5.2 指定时间内的数据强制刷写到hbase
            mutator.flush();
        } catch (Exception ex) {
            logger.error("写入到hbase失败：" + ex.getMessage());
        }
    }

    //4.重写close方法
    @Override
    public void close() throws Exception {
        //4.1 关闭hbase 表和连接资源
        if (mutator != null) mutator.close();
        if (!conn.isClosed()) conn.close();
    }

    //6. 实现 setDataSourcePut 方法

    /**
     * 每条对象生成一个 put
     * 1.表名 2.rowkey 3.列簇  4.列名和列值
     *
     * @param itcastDataObj
     * @return
     */
    private Put setDataSourcePut(ItcastDataObj itcastDataObj) {
        //6.1 如何设计rowkey VIN+时间戳翻转
        String rowkey = itcastDataObj.getVin() + "_" +
                StringUtil.reverse(itcastDataObj.getTerminalTimeStamp() + "");
        logger.info("当前保存的rowkey：" + rowkey);
        //6.3 通过 rowkey 实例化 put
        Put put = new Put(Bytes.toBytes(rowkey));
        //6.2 定义列簇的名称
        String cf = "cf";
        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("vin"), Bytes.toBytes(itcastDataObj.getVin()));
        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("terminalTime"), Bytes.toBytes(itcastDataObj.getTerminalTime()));

        //电量百分比(currentElectricity)、当前电量(remainPower)、百公里油耗(fuelConsumption100km)、
        // 发动机速度(engineSpeed)、车辆速度(vehicleSpeed)
        if (itcastDataObj.getCurrentElectricity() != -999999D) {
            put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("currentElectricity"), Bytes.toBytes(itcastDataObj.getCurrentElectricity() + ""));
        }
        if (itcastDataObj.getRemainPower() != -999999D) {
            put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("remainPower"), Bytes.toBytes(itcastDataObj.getRemainPower() + ""));
        }
        if (StringUtils.isNotEmpty(itcastDataObj.getFuelConsumption100km())) {
            put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("fuelConsumption100km"), Bytes.toBytes(itcastDataObj.getFuelConsumption100km()));
        }
        if (StringUtils.isNotEmpty(itcastDataObj.getEngineSpeed())) {
            put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("engineSpeed"), Bytes.toBytes(itcastDataObj.getEngineSpeed()));
        }
        if (itcastDataObj.getVehicleSpeed() != -999999D) {
            put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("vehicleSpeed"), Bytes.toBytes(itcastDataObj.getVehicleSpeed() + ""));
        }
        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("processTime"), Bytes.toBytes(DateUtil.getCurrentDateTime()));
        //6.4 将所有的字段添加到put的字段中
        return put;
    }
}
